package cn.com.guage.flink.window;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import cn.com.guage.flink.domain.LineSplitter;

/**
 * @version:1.0.1
 * @Description: WindowAll用法
 * @author: yangdechao
 * @date: datedate 2021年11月23日 下午3:33:22
 */
public class WindowAll {

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<String> data = env.socketTextStream("192.168.0.128", 9528);
		SingleOutputStreamOperator<Tuple2<Long, String>> sum = data.flatMap(new LineSplitter()).windowAll(TumblingProcessingTimeWindows.of(Time.seconds(5))).sum(0);
		sum.print();
		env.execute("WindowAll");
	}
}
